package com.qdu.topn;

import java.io.IOException;
import java.util.TreeMap;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TopnMapper extends Mapper<LongWritable, Text, Text, Text>{
    private TreeMap<Integer, String> repToRecordMap = new TreeMap<Integer, String>();
    String map_key="map_key";
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)throws IOException, InterruptedException {
        String line = value.toString();
        if (line.contains("id")) {
            return;
        }
        String[] nums = line.split(",");
        String num=nums[8];
        System.out.println(num+"*************");
            repToRecordMap.put(Integer.parseInt(num),value.toString());
            if(repToRecordMap.size()>10) {
                repToRecordMap.remove(repToRecordMap.firstKey());
            }
    }
    protected void cleanup (Mapper<LongWritable, Text, Text, Text>.Context context) {
        for(String i : repToRecordMap.values()) {
            try {
                context.write(new Text(map_key), new Text(i));
            }catch(Exception e) {
                e.printStackTrace();
            }

        }
    }

}
